package websocket;

import info.bitrich.xchangestream.bitstamp.BitstampStreamingExchange;
import info.bitrich.xchangestream.core.StreamingExchange;
import info.bitrich.xchangestream.core.StreamingExchangeFactory;
import io.reactivex.disposables.Disposable;
import org.knowm.xchange.currency.CurrencyPair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by lujango on 2018/7/25.
 */
public class Bitstamp {
  public static Logger LOG = LoggerFactory.getLogger(Bitstamp.class);

  public static void main(String[] args) throws Exception {
    StreamingExchange exchange = StreamingExchangeFactory.INSTANCE.createExchange(BitstampStreamingExchange.class.getName());

// Connect to the Exchange WebSocket API. Blocking wait for the connection.
    exchange.connect().blockingAwait();

    System.out.println("isalive :"+exchange.isAlive());
// Subscribe to live trades update.
    exchange.getStreamingMarketDataService()
        .getTrades(CurrencyPair.BTC_USD)
        .subscribe(trade -> {
          System.out.println("Incoming trade: {}"+trade);
        }, throwable -> {
          LOG.error("Error in subscribing trades.", throwable);
        });



  //  Thread.sleep(100000);
// Subscribe order book data with the reference to the subscription.
    Disposable subscription = exchange.getStreamingMarketDataService()
        .getOrderBook(CurrencyPair.BTC_USD)
        .subscribe(orderBook -> {
          System.out.println(orderBook.toString());
        });

Thread.sleep(10000000);
/*
// Unsubscribe from data order book.
    subscription.dispose();

// Disconnect from exchange (non-blocking)
    exchange.disconnect().subscribe(() -> LOG.info("Disconnected from the Exchange"));*/
  }
}
